#include "common.hpp"

#include <thread>
#include <mutex>
#include <deque>
#include <atomic>
#include <chrono>
#include <sstream>
#include <condition_variable>

using namespace std;

class Source
{
public:
    Source()
    {
        total_.store(0);
    }

    void Push(std::string msg)
    {
        unique_lock<mutex> guard(mtx_);
        queue_.push_back(msg);
        total_.store(total_.load() + 1);
        cond_.notify_all();
    }

    void Notify()
    {
        unique_lock<mutex> guard(mtx_);
        cond_.notify_all();
    }

    bool FinishFetch()
    {
        unique_lock<mutex> guard(mtx_cnt_);
        return cnt_ == 16;
    }

    bool Fetch(std::string &msg)
    {
        if (FinishFetch())
            return false;
        {
            unique_lock<mutex> guard(mtx_);
            cond_.wait(guard, [&]() -> bool
                       { return !queue_.empty(); });
            msg = queue_.front();
            queue_.pop_front();
            mtx_.unlock();
        }
        {
            unique_lock<mutex> guard_cnt(mtx_cnt_);
            cnt_++;
        }
        return true;
    }

    atomic<int> total_;

    mutex mtx_;
    condition_variable cond_;
    deque<std::string> queue_;

    int cnt_ = 0;
    mutex mtx_cnt_;
};

class Base
{
public:
    Base(Source *src, int id)
        : id_(id),
          src_(src)
    {
        thd_ = new thread(&Base::Run, this);
    }

    virtual ~Base()
    {
        if (thd_ && thd_->joinable())
            thd_->join();
    }

    void Join()
    {
        if (thd_)
            thd_->join();
    }

    virtual void Run() = 0;

    int id_ = 0;

    thread *thd_ = NULL;

    Source *src_ = NULL;
};

class Producer : public Base
{
public:
    Producer(Source *src, int id)
        : Base(src, id) {}

    virtual ~Producer()
    {
    }

    void Run()
    {
        for (int i = 0; i < 8; i++)
        {
            stringstream sss;
            sss << "p" << id_ << "_" << i;
            src_->Push(sss.str());
            // printf("produce[%d] %d", id_, d);
            this_thread::sleep_for(chrono::seconds(2));
        }
        printf("produce[%d] push finish\n", id_);
        while (!src_->FinishFetch())
        {
            src_->Notify();
            this_thread::sleep_for(chrono::seconds(2));
        }
        printf("produce[%d] finish\n", id_);
    }
};

class Consumer : public Base
{
public:
    Consumer(Source *src, int id)
        : Base(src, id)
    {
    }

    virtual ~Consumer()
    {
    }

    void Run()
    {
        string msg;
        while (src_->Fetch(msg))
        {
            printf("consume[%d] %s\n", id_, msg.c_str());
            this_thread::sleep_for(chrono::seconds(3));
        }
        printf("consume[%d] finish\n", id_);
    }
};

int main(int argc, char **argv)
{
    Source src;
    Producer pro1(&src, 1);
    Producer pro2(&src, 2);
    Consumer con1(&src, 1);
    Consumer con2(&src, 2);
    Consumer con3(&src, 3);
    pro1.Join();
    pro2.Join();
    con1.Join();
    con2.Join();
    con3.Join();
    printf("total %d %d\n", src.cnt_, src.total_.load());
    return 1;
}